package com.jt;

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.PullStatus;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

public class PullConsumerTest {

    public static volatile boolean running = true;

    public static void main(String[] args) throws MQClientException {
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("zero-group-01-c");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.start();
        try {
            MessageQueue mq = new MessageQueue();
            mq.setQueueId(0);
            mq.setTopic("TopicTest");
            mq.setBrokerName("zero-virtual-machine");
            long offset = 26;
                PullResult pullResult = consumer.pull(mq, "*", offset, 32);
                if (pullResult.getPullStatus().equals(PullStatus.FOUND)) {
                    System.out.printf("%s%n", pullResult.getMsgFoundList());
                    for(MessageExt item: pullResult.getMsgFoundList()){
                        //item.getBody();
                        String s = new String(item.getBody());

                        System.out.printf("%s%n", s);
                    }
                    consumer.updateConsumeOffset(mq, pullResult.getNextBeginOffset());
                }


        } catch (Exception e) {
            e.printStackTrace();
        }
        consumer.shutdown();
    }
}
